package com.zytech.zspider.eventbus.support;

import com.zytech.modules.lang.Langs;
import com.zytech.modules.lang.Requires;
import com.zytech.zspider.eventbus.EventListener;
import com.zytech.zspider.logger.Loggers;

import java.util.Set;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

/**
 * @author siuming
 */
public abstract class AutoRetryEventListenerContainer implements EventListenerContainer {

    protected static final long MAX_DELAY_TIME = 128;
    protected static final long FIRST_DELAY_TIME = 0;

    private final ScheduledExecutorService scheduledService;
    private final Set<Class<? extends Exception>> retriableExceptionTypes;

    /**
     * @param scheduledService the scheduledService use to retry add listeners.
     * @param retriableExceptionTypes   the exceptionTypes to retry.
     */
    protected AutoRetryEventListenerContainer(ScheduledExecutorService scheduledService,
                                              Set<Class<? extends Exception>> retriableExceptionTypes) {
        Requires.notNull(scheduledService, "scheduledService must not be null.");
        Requires.notNull(retriableExceptionTypes, "retriableExceptionTypes must not be null.");
        this.scheduledService = scheduledService;
        this.retriableExceptionTypes = retriableExceptionTypes;
    }

    @Override
    public final void addTopicListener(String tname, EventListener listener) {
        TaskClock clock = new TaskClock(MAX_DELAY_TIME, FIRST_DELAY_TIME);
        Runnable task = new AddTopicListenerTask(tname, listener, clock);
        scheduledService.schedule(task, clock.firstDelayTime(), TimeUnit.SECONDS);
    }

    class AddTopicListenerTask implements Runnable {

        final String tname;
        final EventListener listener;
        final TaskClock clock;

        AddTopicListenerTask(String tname, EventListener listener, TaskClock clock) {
            this.tname = tname;
            this.listener = listener;
            this.clock = clock;
        }

        @Override
        public void run() {
            try {
                addTopicListenerInternal(tname, listener);
            } catch (Exception e) {
                if (isRetriableException(e)) {
                    retryAddTopicListener(this, clock.nextDelayTime());
                    return;
                }

                throw Langs.toUncheck(e);
            }
        }
    }

    private void retryAddTopicListener(AddTopicListenerTask task, long nextDelayTime) {
        Loggers.me().warn(
                getClass(),
                "it seems some error occurs. retry add topic listener after {} seconds.",
                nextDelayTime
        );
        scheduledService.schedule(task, nextDelayTime, TimeUnit.SECONDS);
    }

    private boolean isRetriableException(Exception e) {
        return retriableExceptionTypes.contains(e.getClass());
    }

    /**
     * @param tname
     * @param listener
     * @throws Exception
     */
    protected abstract void addTopicListenerInternal(String tname, EventListener listener) throws Exception;

    @Override
    public final void addQueueListener(String qname, EventListener listener) {
        TaskClock clock = new TaskClock(MAX_DELAY_TIME, FIRST_DELAY_TIME);
        Runnable task = new AddQueueListenerTask(qname, listener, clock);
        scheduledService.schedule(task, clock.firstDelayTime(), TimeUnit.SECONDS);
    }

    class AddQueueListenerTask implements Runnable {

        final String qname;
        final EventListener listener;
        final TaskClock clock;

        AddQueueListenerTask(String qname, EventListener listener, TaskClock clock) {
            this.qname = qname;
            this.listener = listener;
            this.clock = clock;
        }

        @Override
        public void run() {
            try {

                addQueueListenerInternal(qname, listener);
            } catch (Exception e) {
                if (isRetriableException(e)) {
                    retryAddQueueListener(this, clock.nextDelayTime());
                    return;
                }

                throw Langs.toUncheck(e);
            }
        }
    }

    private void retryAddQueueListener(AddQueueListenerTask task, long nextDelayTime) {
        Loggers.me().warn(
                getClass(),
                "it seems some error occurs. will retry add queue listener after {} seconds.",
                nextDelayTime
        );

        scheduledService.schedule(task, nextDelayTime, TimeUnit.SECONDS);
    }

    /**
     * @param qname
     * @param listener
     * @throws Exception
     */
    protected abstract void addQueueListenerInternal(String qname, EventListener listener) throws Exception;

    class TaskClock {
        final long maxDelayTime;
        final long firstDelayTime;
        final AtomicLong current;

        TaskClock(long maxDelayTime, long firstDelayTime) {
            this.maxDelayTime = maxDelayTime;
            this.firstDelayTime = firstDelayTime;
            this.current = new AtomicLong(firstDelayTime);
        }

        long firstDelayTime() {
            return firstDelayTime;
        }

        long nextDelayTime() {
            long prevDelayTime = current.get() == 0 ? 1L : current.get();
            current.set(prevDelayTime * 2);
            if (current.get() > maxDelayTime) {
                current.set(firstDelayTime);
                return firstDelayTime;
            }
            return current.get();
        }
    }
}
